﻿using System;
using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;
using DelayedQueue.Models;
using DelayedQueue.TimingWheel;

namespace DelayedQueue.Kafka
{
    /// <summary>
    /// Kafka消费者，负责接收延时任务并添加到时间轮
    /// </summary>
    public class KafkaConsumer : IDisposable
    {
        private readonly IConsumer<string, string> _consumer;
        private readonly string _topic;
        private readonly TimingWheel.TimingWheel _timingWheel;
        private readonly Data.MySQLPersistence _persistence;
        private CancellationTokenSource _cts;
        private bool _isConsuming;
        private bool _isDisposed;

        public KafkaConsumer(
            string bootstrapServers,
            string topic,
            string groupId,
            TimingWheel.TimingWheel timingWheel,
            Data.MySQLPersistence persistence)
        {
            if (string.IsNullOrEmpty(bootstrapServers))
                throw new ArgumentException("Kafka服务器地址不能为空", nameof(bootstrapServers));

            if (string.IsNullOrEmpty(topic))
                throw new ArgumentException("主题名称不能为空", nameof(topic));

            if (string.IsNullOrEmpty(groupId))
                throw new ArgumentException("消费者组ID不能为空", nameof(groupId));

            _topic = topic;
            _timingWheel = timingWheel ?? throw new ArgumentNullException(nameof(timingWheel));
            _persistence = persistence ?? throw new ArgumentNullException(nameof(persistence));

            var config = new ConsumerConfig
            {
                BootstrapServers = bootstrapServers,
                GroupId = groupId,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnableAutoCommit = false  // 禁用自动提交，手动控制偏移量
            };

            _consumer = new ConsumerBuilder<string, string>(config).Build();
        }

        /// <summary>
        /// 开始消费消息
        /// </summary>
        public void StartConsuming()
        {
            if (_isConsuming)
                return;

            _isConsuming = true;
            _cts = new CancellationTokenSource();

            _consumer.Subscribe(_topic);

            // 启动消费线程
            Task.Run(() => ConsumeLoop(_cts.Token));
        }

        /// <summary>
        /// 消费循环
        /// </summary>
        private void ConsumeLoop(CancellationToken cancellationToken)
        {
            try
            {
                while (!cancellationToken.IsCancellationRequested)
                {
                    try
                    {
                        var consumeResult = _consumer.Consume(cancellationToken);

                        if (consumeResult.Message != null)
                        {
                            try
                            {
                                // 解析消息
                                var task = DelayedTask.FromJson(consumeResult.Message.Value);

                                // 保存到数据库
                                _persistence.SaveTask(task);

                                // 添加到时间轮
                                _timingWheel.AddTask(task);

                                Console.WriteLine($"接收并添加任务: {task.TaskId}, 执行时间: {task.ExecuteTimeUtc:yyyy-MM-dd HH:mm:ss}");

                                // 手动提交偏移量
                                _consumer.Commit(consumeResult);
                            }
                            catch (Exception ex)
                            {
                                Console.WriteLine($"处理消息失败: {ex.Message}");
                            }
                        }
                    }
                    catch (ConsumeException ex)
                    {
                        Console.WriteLine($"消费异常: {ex.Error.Reason}");
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // 预期的取消异常，无需处理
            }
            catch (Exception ex)
            {
                Console.WriteLine($"消费循环异常: {ex.Message}");
            }
            finally
            {
                _consumer.Close();
            }
        }

        /// <summary>
        /// 停止消费
        /// </summary>
        public void StopConsuming()
        {
            if (!_isConsuming)
                return;

            _isConsuming = false;
            _cts?.Cancel();
            _cts?.Dispose();
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (_isDisposed) return;

            if (disposing)
            {
                StopConsuming();
                _consumer?.Dispose();
            }

            _isDisposed = true;
        }

        ~KafkaConsumer()
        {
            Dispose(false);
        }
    }
}
